package com.taldh.netty.io.reactor;


import com.taldh.netty.io.NioDemoConfig;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;


class MultiThreadEchoServerReactor {
    ServerSocketChannel serverSocket;
    //selectors集合,引入多个selector选择器
    //引入多个子反应器
    SubReactor[] subReactors;
    MainReactor mainReactor;
    private final int processors = Runtime.getRuntime().availableProcessors();

    MultiThreadEchoServerReactor() throws IOException {
        subReactors = new SubReactor[processors];
        for (int i = 0; i < processors; i++) {
            subReactors[i] = new SubReactor(Selector.open());
        }
        mainReactor = new MainReactor(Selector.open(), subReactors);
    }


    private void startService() {
        // 一子反应器对应一条线程
        ExecutorService pool = Executors.newFixedThreadPool(processors + 1);
        pool.execute(mainReactor);
        for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) {
            pool.execute(subReactors[i]);
        }
    }

    class MainReactor implements Runnable {

        AtomicInteger next = new AtomicInteger(0);

        //每条线程负责一个选择器的查询
        final Selector selector;

        public MainReactor(Selector selector,SubReactor[] subReactors) {
            this.selector = selector;
            //初始化多个selector选择器
            try {
                serverSocket = ServerSocketChannel.open();
                InetSocketAddress address =
                        new InetSocketAddress(NioDemoConfig.SOCKET_SERVER_IP,
                                NioDemoConfig.SOCKET_SERVER_PORT);
                serverSocket.socket().bind(address);
                //非阻塞
                serverSocket.configureBlocking(false);

                //第一个selector,负责监控新连接事件
                SelectionKey sk = serverSocket.register(this.selector, SelectionKey.OP_ACCEPT);
                //附加新连接处理handler处理器到SelectionKey（选择键）
                sk.attach(new AcceptorHandler(subReactors));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            try {
                for(;;) {
                    if (selector.select() > 0) {
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        for (SelectionKey selectionKey : selectionKeys) {
                            dispatch(selectionKey);
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private void dispatch(SelectionKey selectionKey) {
            Runnable handler = (Runnable) selectionKey.attachment();
            //调用之前attach绑定到选择键的handler处理器对象
            if (handler != null) {
                handler.run();
            }
        }

        // Handler:新连接处理器
        class AcceptorHandler implements Runnable {

            private final SubReactor[] subReactors;

            public AcceptorHandler(SubReactor[] subReactor) {
                this.subReactors = subReactor;
            }

            public void run() {
                try {
                    SocketChannel channel = serverSocket.accept();
                    if (channel != null)
                        new MultiThreadEchoHandler(subReactors[next.get()].getSelector(), channel);
                } catch (IOException e) {
                    e.printStackTrace();
                }
                if (next.incrementAndGet() == subReactors.length) {
                    next.set(0);
                }
            }
        }
    }

    //反应器
    static class SubReactor implements Runnable {
        //每条线程负责一个选择器的查询
        final Selector selector;

        public SubReactor(Selector selector) {
            this.selector = selector;
        }

        public Selector getSelector() {
            return selector;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    if (selector.select() > 0) {
                        Set<SelectionKey> keySet = selector.selectedKeys();
                        for (SelectionKey sk : keySet) {
                            //Reactor负责dispatch收到的事件
                            dispatch(sk);
                        }
                        keySet.clear();
                    }
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }


        void dispatch(SelectionKey sk) {
            Runnable handler = (Runnable) sk.attachment();
            //调用之前attach绑定到选择键的handler处理器对象
            if (handler != null) {
                handler.run();
            }
        }
    }


    public static void main(String[] args) throws IOException {
        MultiThreadEchoServerReactor server =
                new MultiThreadEchoServerReactor();
        server.startService();
    }

}
